package com.test.rxjava2;

import io.reactivex.Observable;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.annotations.NonNull;
import io.reactivex.disposables.Disposable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class Demo01 {
    private static final Logger Log = LoggerFactory.getLogger(Demo01.class);
    private static final String TAG = "Demo01";
    public static void main(String[] args) {
        // 第一步：初始化Observable
        Observable.create((ObservableOnSubscribe<Integer>) e -> {
            Log.error(TAG, "Observable emit 1" + "\n");
            e.onNext(1);
            Log.error(TAG, "Observable emit 2" + "\n");
            e.onNext(2);
            Log.error(TAG, "Observable emit 3" + "\n");
            e.onNext(3);
            e.onComplete();
            Log.error(TAG, "Observable emit 4" + "\n" );
            e.onNext(4);
        }).subscribe(new Observer<Integer>() { // 第三步：订阅

            // 第二步：初始化Observer
            private int i;
            private Disposable mDisposable;

            @Override
            public void onSubscribe(@NonNull Disposable d) {
                mDisposable = d;
            }

            @Override
            public void onNext(@NonNull Integer integer) {
                i++;
                if (i == 2) {
                    // 在RxJava 2.x 中，新增的Disposable可以做到切断的操作，让Observer观察者不再接收上游事件
                    mDisposable.dispose();
                }
            }

            @Override
            public void onError(@NonNull Throwable e) {
                Log.error(TAG, "onError : value : " + e.getMessage() + "\n" );
            }

            @Override
            public void onComplete() {
                Log.error(TAG, "onComplete" + "\n" );
            }
        });
    }
}
